package drds.data_migrate.task_group.task._runnable_;


import drds.common.Author;
import drds.data_migrate.Authors;
import drds.data_migrate.task_group.task.data_exchange.exchanger.Consume;
import drds.data_migrate.task_group.task.run.WriteTask;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Validate;

@Slf4j
@Author(name = Authors.LI_YANG)
public class WriteRunnable extends AbstractRunnable implements Runnable
{


    @Setter
    @Getter
    private Consume consume;
    @Setter
    @Getter
    private WriteTask writeTask;
    @Setter
    @Getter
    private TaskConfig taskConfig;

    public void setConsume(Consume consume)
    {
        this.consume = consume;
    }


    public void run()
    {
        Validate.isTrue(this.consume != null);

        writeTask = new WriteTask(taskConfig);


        try
        {

            //
            log.debug("wrapper write starts to do init ...");


            writeTask.init();


            log.debug("wrapper write starts to do prepare ...");

            writeTask.prepare();

            log.debug("wrapper write starts to write ...");


            writeTask.startWrite(consume);


            log.debug("wrapper write starts to do post ...");

            writeTask.post();


            writeTask.destroy();
            super.markSuccess();
        } catch (Throwable e)
        {
            log.error("AbstractWriter Runner Received Exceptions:", e);
            super.markFail(e);
        } finally
        {
            log.debug("wrapper write starts to do destroy ...");
            writeTask.destroy();

        }
    }


    public boolean supportFailOver()
    {
        String writeMode = null;
        return "replace".equalsIgnoreCase(writeMode);
    }


    public void shutdown()
    {
        consume.shutdown();
    }
}
